/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.adaptive.selection;

import com.facebook.airlift.log.Logger;
import com.facebook.presto.adaptive.AdaptiveMetadataManager;
import com.facebook.presto.adaptive.AdaptiveUtil;
import com.facebook.presto.spi.ColumnHandle;
import com.facebook.presto.spi.ConnectorSession;
import com.facebook.presto.spi.TableHandle;
import com.facebook.presto.spi.adaptive.CatalogEngine;
import com.facebook.presto.spi.adaptive.ConnectorAdaptiveOptimizer;
import com.facebook.presto.spi.adaptive.OptimizerType;
import com.facebook.presto.spi.adaptive.metadata.presto.AdaptivePrestoObject;
import com.facebook.presto.spi.plan.FilterNode;
import com.facebook.presto.spi.plan.PlanNode;
import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
import com.facebook.presto.spi.plan.PlanVisitor;
import com.facebook.presto.spi.plan.TableScanNode;
import com.facebook.presto.spi.predicate.TupleDomain;
import com.facebook.presto.spi.relation.VariableReferenceExpression;
import com.google.common.collect.ImmutableList;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

import static com.facebook.presto.spi.adaptive.AdaptiveConstant.TABLE_PROPERTY_SEPARATOR;
import static java.util.Objects.requireNonNull;

/**
 * @author le.Mi
 * WideTableSelector: a RBO adptive selector
 * <p>
 * metadata:
 * query :
 * result : => transform to hbase.
 */
public class LessColumnSelector
        extends RBOAdaptiveSelector
{
    private static final Logger log = Logger.get(LessColumnSelector.class);

    private String table_property_separator = TABLE_PROPERTY_SEPARATOR;

    List<ConnectorAdaptiveOptimizer> connectorAdaptiveOptimizers;

    @Override
    public PlanNode adaptiveSelect(PlanNode maxSubplan,
                                   ConnectorSession session,
                                   PlanNodeIdAllocator idAllocator,
                                   List<ConnectorAdaptiveOptimizer> connectorAdaptiveOptimizers)
    {
        this.connectorAdaptiveOptimizers = connectorAdaptiveOptimizers;
        return maxSubplan.accept(new LessColumnSelector.Visitor(session, idAllocator), null);
    }

    private class Visitor
            extends PlanVisitor<PlanNode, Void>
    {
        private final ConnectorSession session;
        private final PlanNodeIdAllocator idAllocator;

        public Visitor(ConnectorSession session, PlanNodeIdAllocator idAllocator)
        {
            this.session = requireNonNull(session, "session is null");
            this.idAllocator = requireNonNull(idAllocator, "idAllocator is null");
        }

        @Override
        public PlanNode visitPlan(PlanNode node, Void context)
        {
            ImmutableList.Builder<PlanNode> children = ImmutableList.builder();
            boolean changed = false;
            for (PlanNode child : node.getSources()) {
                PlanNode newChild = child.accept(this, null);
                if (newChild != child) {
                    changed = true;
                }
                children.add(newChild);
            }

            if (!changed) {
                return node;
            }
            return node.replaceChildren(children.build());
        }

        @Override
        public PlanNode visitFilter(FilterNode node, Void context)
        {
            if (!(node.getSource() instanceof TableScanNode)) {
                return node;
            }
            TableScanNode tableScanNode = (TableScanNode) node.getSource();

            TableHandle tableHandle = ((TableScanNode) node.getSource()).getTable();
            TupleDomain<ColumnHandle> tupleDomain = tableHandle.getLayout().get().getTupleDomain();

            List<ColumnHandle> domainColumns = new ArrayList<>();

            // fill all columns and select columns with domain
            for (VariableReferenceExpression variable : tableScanNode.getOutputVariables()) {
                ColumnHandle column = tableScanNode.getAssignments().get(variable);

                if (tupleDomain.getDomains().get().get(column) != null) {
                    domainColumns.add(column);
                }
            }

            // select execute engine and optimize
            FilterNode newNode = (FilterNode) adaptiveSelectAndOptimize(node, tableHandle, domainColumns, idAllocator);

            return newNode;
        }
    }

    private PlanNode adaptiveSelectAndOptimize(PlanNode node,
                                               TableHandle tableHandle,
                                               List<ColumnHandle> domainColumns,
                                               PlanNodeIdAllocator idAllocator)
    {
        CatalogEngine catalogEngine = selectPushDownEngine(tableHandle, domainColumns);
        if (catalogEngine == null) {
            return node;
        }

        PlanNode newNode = node;
        if (catalogEngine.equals(CatalogEngine.HBASE)) {
            List<ConnectorAdaptiveOptimizer> optimizers = filterAdaptiveOptimizers(this.connectorAdaptiveOptimizers, CatalogEngine.HBASE, OptimizerType.WIDETABLE);
            if (optimizers == null) {
                return node;
            }

            AdaptivePrestoObject adaptiveObject = AdaptiveUtil.getMetadataManager()
                    .getAdaptivePrestoObject(CatalogEngine.HBASE.toString(), null,
                            tableHandle.getConnectorHandle().getSchemaTableName().getTableName());

            for (ConnectorAdaptiveOptimizer optimizer : optimizers) {
                newNode = optimizer.adaptiveOptimize(newNode, tableHandle, idAllocator, adaptiveObject);
            }
        }
        return newNode;
    }

    /**
     * select push down engine.
     * two conditions:
     * 1. the engine has this table.
     * 2. the engine is satisfied push-down rules.
     */
    private CatalogEngine selectPushDownEngine(TableHandle tableHandle, List<ColumnHandle> domainColumns)
    {
        String tableName = tableHandle.getConnectorHandle().getSchemaTableName().getTableName();

        AdaptiveMetadataManager metadataManager = AdaptiveUtil.getMetadataManager();
        List<String> candidateCatalog = metadataManager.getCatalogListByTableName(tableName);

        if (candidateCatalog.contains(CatalogEngine.HBASE.toString()) && isSatisfiedWideTable(tableName, domainColumns)) {
            log.info("Adaptive selector - Select Engine, MatchRowkeySelector selects hbase engine.");
            return CatalogEngine.HBASE;
        }
        return null;
    }


    /**
     *
     * @param tableName
     * @param domainColumns
     * @return
     */
    private boolean isSatisfiedWideTable(String tableName, List<ColumnHandle> domainColumns)
    {
        AdaptiveMetadataManager metadataManager = AdaptiveUtil.getMetadataManager();
        AdaptivePrestoObject adaptiveObject = metadataManager
                .getAdaptivePrestoObject(CatalogEngine.HBASE.toString(), null, tableName);
        String rowkeyPattern = adaptiveObject.getPrestoTable().getHbaseRowkey();

        List<String> domainColumnNameList = new ArrayList<>();
        if (!domainColumns.isEmpty()) {
            domainColumnNameList = domainColumns.stream().map(ColumnHandle::getColumnName).collect(Collectors.toList());
        }

        // rowkeyPattern: salt/columnName-length/primaryKey-length

        if (rowkeyPattern == null || rowkeyPattern.isEmpty()) {
            return false;
        }

        String[] patternParts = rowkeyPattern.split(table_property_separator);

        String columnNameAndLength = patternParts[1];
        String columnName = (columnNameAndLength.split("-"))[0];

        return domainColumnNameList.contains(columnName);
    }
}
